--
--  Copyright 2022 The OpenIslands Authors. All Rights Reserved.
--
--  Licensed under the Apache License, Version 2.0 (the "License");
--  you may not use this file except in compliance with the License.
--  You may obtain a copy of the License at
--
--      http://www.apache.org/licenses/LICENSE-2.0
--
--  Unless required by applicable law or agreed to in writing, software
--  distributed under the License is distributed on an "AS IS" BASIS,
--  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
--  See the License for the specific language governing permissions and
--  limitations under the License.
--
local _M = {
    _VERSION = '0.1'
}

local ngx = ngx
local utils = require "utils"
local config = require "config"
local dns_utils = require "dns_utils"
local producer = require "resty.kafka.producer"
local log = require("log")
local table = table
local tonumber = tonumber

function _M.publish(routing_info, headers, data)
    local broker_config, err = _M.get_broker_config()
    if not broker_config then
        return nil, err
    end
    -- timeout in milliseconds
    local p = producer:new(broker_config, {keepalive_timeout=120, keepalive_size=50})
    -- local p = producer:new(broker_config, {keepalive_timeout=120, keepalive_size=50, producer_type="async", request_timeout=3000, max_retry=100, retry_backoff=100, batch_size=1048576, batch_num=50})
    local packet_properties = utils.json_loads(headers["properties"])
    local packet_format = packet_properties["format"]
    local body
    if packet_format == "PLAIN" then
        body = utils.string_to_hex(data)
    else
        body = data
    end
    local message = {headers=headers, body=body}
    local dest_topic = routing_info["dest-receive-topic"]
    log.debug("try publish message to ", dest_topic)
    local message_str = utils.json_dumps(message)
    local ok, err = p:send(dest_topic, nil, message_str)
    -- ok is number when sync and is true when async
    if not ok then
        return nil, dest_topic..": "..err
    end
    log.debug("publish message to ", dest_topic)
    return ok, "success"
end

function _M.get_broker_config()
    local broker_hosts = utils.string_split(config.kafka["servers"], ",")
    local broker_config = {}
    for i, v in pairs(broker_hosts) do
        local items = utils.string_split(v, ":")
        local ip, err = dns_utils.get_ip_address(items[1])
        if ip == nil then
            err = "get kafka broker ip using domain: "..items[1].." by dns failed"
            log.error(err)
            return nil, err
        end
        table.insert(broker_config, {host=ip, port=tonumber(items[2])})
    end
    return broker_config, err
end

return _M